package org.hadoop.writable;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;
public class LineCharCountMR extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        int run = ToolRunner.run(new LineCharCountMR(), args);
        System.exit(run);
    }
    @Override
    public int run(String[] args) throws Exception {
        if (args.length < 2) {
            System.out.println("参数错误，使用方法：LineCharCountMR <Input> <Output>");
            ToolRunner.printGenericCommandUsage(System.out);
            return 1;
        }
        Configuration config = getConf();
        FileSystem fs = FileSystem.get(config);
        Path dest = new Path(args[1]);
        if (fs.exists(dest)) {
            fs.delete(dest, true);
        }
        Job job = Job.getInstance(config, "LineChar");
        job.setJarByClass(getClass());
        job.setMapperClass(LineMapper.class);
        job.setMapOutputKeyClass(LineCharCountWritable.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setReducerClass(LineReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, dest);
        //设置Reducer的个数
        job.setNumReduceTasks(2);
        job.setPartitionerClass(LineCharCountPartitioner.class);
        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    }
    //注意最后一个参数为NullWritable，可以理解为Null
    public static class LineMapper extends Mapper<LongWritable, Text, LineCharCountWritable, NullWritable> {
        private LineCharCountWritable countWritable = new LineCharCountWritable();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (StringUtils.isBlank(line)) {
                return;
            }
            Integer charCount = line.length();
            countWritable.setLine(line);
            countWritable.setCount(charCount);
            context.write(countWritable, NullWritable.get());
        }
    }
    public static class LineReducer extends Reducer<LineCharCountWritable, NullWritable, Text, NullWritable> {
        private Text text = new Text();
        @Override
        protected void reduce(LineCharCountWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            text.set(key.getLine() + "\t" + key.getCount());
            context.write(text, NullWritable.get());
        }
    }
}
